package com.hl233.framework.utils;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;

/**
 * kafka处理类
 * @author 张振强15065206763
 * @create 2024-09-03
 */
@Component
public class KafkaUtil<T> {
    private static final Logger log = LoggerFactory.getLogger(KafkaUtil.class);

    private final KafkaTemplate kafkaTemplate;

    @Autowired
    public KafkaUtil(KafkaTemplate kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    /**
     * 发送kafka 消息
     *
     * @param topic kafka topic name
     * @param t     java object
     */
    @SuppressWarnings("unchecked")
    public void sendTask(String topic, T t) {
        JSONObject jsonDto = (JSONObject) JSON.toJSON(t);
        ListenableFuture future = kafkaTemplate.send(topic, jsonDto.toJSONString());
        future.addCallback(o -> {},
                throwable -> log.info("error kafka"));
    }

    /**
     * 发送kafka 消息
     *
     * @param topic  topic
     * @param detail data & string
     */
    @SuppressWarnings("unchecked")
    public void send(String topic, String detail) {
        ListenableFuture future = kafkaTemplate.send(topic, detail);
        future.addCallback(o -> System.out.println("send-消息发送成功：" + detail),
                throwable -> log.info("error kafka"));
    }
}
